package com.millstein.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.millstein.realtime.app.base.BaseAppV2;
import com.millstein.realtime.bean.TrafficPageViewBean;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.CommonUtil;
import com.millstein.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Map;

public class Dws_02_TrafficVcChArIsNewPageViewWindow extends BaseAppV2 {

    public static void main(String[] args) {
        new Dws_02_TrafficVcChArIsNewPageViewWindow().init(
                8002,
                1,
                "Dws_02_TrafficVcChArIsNewPageViewWindow",
                Constants.TOPIC_DWD_TRAFFIC_PAGE,
                Constants.TOPIC_DWD_TRAFFIC_UV_DETAIL,
                Constants.TOPIC_DWD_TRAFFIC_UJ_DETAIL
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     * @param env 执行环境对象
     * @param streamSourceInfo 待处理的数据流
     */
    @Override
    protected void handle(
            StreamExecutionEnvironment env,
            Map<String, DataStreamSource<String>> streamSourceInfo
    ) {
        // 1.三流union
        DataStream<TrafficPageViewBean> unionStream = parseAndUnionOne(streamSourceInfo);

        // 2.开窗聚合
        SingleOutputStreamOperator<TrafficPageViewBean> resultStream = windowAndAgg(unionStream);

        // 3.将结果写到doris中
        writeToDoris(resultStream);
    }

    /**
     * 读取3个流，并将其union在一起输出
     * @param streamSourceInfo 待读取的流信息
     * @return
     */
    private DataStream<TrafficPageViewBean> parseAndUnionOne(Map<String, DataStreamSource<String>> streamSourceInfo) {
        // 1.将页面访问流转为TrafficPageViewBean流
        // 它的逻辑是有一条页面访问数据，pv记为1，访问时间为duringTime的值，sv在last_page_id为null的时候记为1否则记为0
        SingleOutputStreamOperator<TrafficPageViewBean> pageStream = streamSourceInfo
                .get(Constants.TOPIC_DWD_TRAFFIC_PAGE)
                .map(new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {
                        JSONObject json = JSONObject.parseObject(value);
                        JSONObject common = json.getJSONObject("common");
                        JSONObject page = json.getJSONObject("page");
                        return TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .isNew(common.getString("is_new"))
                                .uvCt(0L)
                                .svCt(page.getString("last_page_id") == null ? 1L : 0L)
                                .pvCt(1L)
                                .durSum(page.getLong("during_time"))
                                .ujCt(0L)
                                .ts(json.getLong("ts"))
                                .build();
                    }
                });

        // 2.将独立访问流转为TrafficPageViewBean流
        // 它的逻辑是有一条独立访客数据，uv记为1
        SingleOutputStreamOperator<TrafficPageViewBean> userStream = streamSourceInfo
                .get(Constants.TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {
                        JSONObject json = JSONObject.parseObject(value);
                        JSONObject common = json.getJSONObject("common");
                        return TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .isNew(common.getString("is_new"))
                                .uvCt(1L)
                                .svCt(0L)
                                .pvCt(0L)
                                .durSum(0L)
                                .ujCt(0L)
                                .ts(json.getLong("ts"))
                                .build();
                    }
                });

        // 3.将访问跳出流转为TrafficPageViewBean流
        // 它的逻辑是有一条跳出数据，uj记为1
        SingleOutputStreamOperator<TrafficPageViewBean> jumpStream = streamSourceInfo
                .get(Constants.TOPIC_DWD_TRAFFIC_UJ_DETAIL)
                .map(new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {
                        JSONObject json = JSONObject.parseObject(value);
                        JSONObject common = json.getJSONObject("common");
                        return TrafficPageViewBean.builder()
                                .vc(common.getString("vc"))
                                .ch(common.getString("ch"))
                                .ar(common.getString("ar"))
                                .isNew(common.getString("is_new"))
                                .uvCt(0L)
                                .svCt(0L)
                                .pvCt(0L)
                                .durSum(0L)
                                .ujCt(1L)
                                .ts(json.getLong("ts"))
                                .build();
                    }
                });

        // 4.将三流union
        return pageStream.union(userStream, jumpStream);
    }

    /**
     * 窗口聚合
     * @param unionStream 3流union后的流
     * @return
     */
    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(DataStream<TrafficPageViewBean> unionStream) {
        return unionStream
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }))
                .keyBy(new KeySelector<TrafficPageViewBean, String>() {
                    @Override
                    public String getKey(TrafficPageViewBean value) throws Exception {
                        return value.getVc() + "_" + value.getCh() + "_" + value.getAr() + "_" + value.getIsNew();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(
                                    TrafficPageViewBean value1,
                                    TrafficPageViewBean value2
                            ) throws Exception {
                                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                                value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                            @Override
                            public void process(
                                    String key,
                                    Context context,
                                    Iterable<TrafficPageViewBean> elements,
                                    Collector<TrafficPageViewBean> out) throws Exception {
                                String startTime = CommonUtil.formatTime(context.window().getStart());
                                String endTime = CommonUtil.formatTime(context.window().getEnd());
                                String curDate = startTime.substring(0, 10);
                                TrafficPageViewBean bean = elements.iterator().next();
                                bean.setStt(startTime);
                                bean.setEdt(endTime);
                                bean.setCurDate(curDate);
                                out.collect(bean);
                            }
                        }
                );
    }

    /**
     * 将结果写到doris中
     * @param resultStream 处理完的数据流
     */
    private void writeToDoris(SingleOutputStreamOperator<TrafficPageViewBean> resultStream) {
        resultStream
                .map(new MapFunction<TrafficPageViewBean, String>() {
                    @Override
                    public String map(TrafficPageViewBean value) throws Exception {
                        SerializeConfig config = new SerializeConfig();
                        config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                        return JSONObject.toJSONString(value, config);
                    }
                }).addSink(FlinkSinkUtil.getDorisSink("gmall.dws_traffic_vc_ch_ar_is_new_page_view_window"));
    }
}


